Cloud Composer で DAG パーサーのログを Cloud Logging に出力したい

結論、Airflow 構成オプション dag_processor_log_target を stdout にする。
#Google Cloud (GCP)
#Cloud Composer
#Cloud Logging
#Apache Airflow
#小ネタ
エノカワ
2024.11.10
こんにちは！エノカワです。
Cloud Composer は、Apache Airflow で構築されたフルマネージドのワークフローオーケストレーションサービスです。

Apache Airflow のオープンソース プロジェクトを基に構築されており、Python プログラミング言語を使用して動作します。
今回の記事では、Cloud Composer で DAG パーサーのログを Cloud Logging に出力する方法について検証した内容をご紹介します。
 やりたいことDAG のトップレベルコードで Airflow 変数にアクセスする際、そのアクセスがパースごとに発生していることを確認したいと考えています。

Cloud Logging を利用して、このプロセスを可視化し、DAG パース時のパフォーマンスに対する影響を評価することを目的としています。
 経緯以下のドキュメントをきっかけに、トップレベルのコードにおける Airflow 変数のアクセスがパースごとに発生していることを Cloud Logging のログで確認しようとしました。

しかし、ログに出力されなかったため、調査を進めた結果、dag_processor_log_target を stdout に設定することでログに出力されることがわかりました。
Airflow Best Practices
Using Airflow Variables yields network calls and database access, so their usage in top-level Python code for DAGs should be avoided as much as possible, as mentioned in the previous chapter, Top level Python Code. If Airflow Variables must be used in top-level DAG code, then their impact on DAG parsing can be mitigated by enabling the experimental cache, configured with a sensible ttl.
Airflow ベストプラクティスドキュメントによると、Airflow 変数をトップレベルのコードで使用することは、ネットワークアクセスやデータベースアクセスを伴うため、できる限り避けるべきだとされています。
 dag_processor_log_target とはdag_processor_log_target は、DAG パーサーが生成するログの出力先を指定するための Airflow の設定オプションです。
 設定値の種類file : 特定のファイルにログを出力します
stdout : 標準出力にログを出力します
この設定をstdoutにすることで、DAG のパース時に発生するログを Cloud Logging に送信することが可能になります。
 環境作成DAG を動かすための Cloud Composer 環境を作成します。

Google Cloud コンソールの「Cloud Composer の環境作成」ページから、test-composerという名前の環境を、東京リージョンで作成しました。

サービスアカウントなどその他の設定はデフォルトのままで作成しました。
 Airflow 変数を設定するAirflow 変数を設定することで、タスク間でデータを共有し、DAG の動作をカスタマイズできます。
今回の検証用に Variable 変数を事前に設定しておきます。

以下のコマンドを使用して、my_variableという変数に my_value を設定しています。
$ gcloud composer environments run test-composer \
    --location asia-northeast1 \
    variables -- set my_variable my_value
 DAG を作成する次に、DAG ファイルを作成します。
ここでは、トップレベルとタスク内でそれぞれ Airflow 変数にアクセスし、その旨がわかるログを出力する DAG を用意します。

これにより、DAG パース時とタスク実行時の両方で変数アクセスのログを確認できます。
variable_access_logging_dag.py
from airflow import DAG
from airflow.models import Variable
from airflow.operators.python import PythonOperator
from datetime import datetime
import logging

def log_variable():
    task_level_var = Variable.get("my_variable", default_var="default_value")
    logging.info(f"Task-level variable access: {task_level_var}")

top_level_var = Variable.get("my_variable", default_var="default_value")
logging.info(f"Top-level variable access: {top_level_var}")

dag = DAG(
    'variable_access_logging_dag_dag',
    default_args={'start_date': datetime(2023, 1, 1)},
    schedule_interval=None,
)

task = PythonOperator(
    task_id='log_variable_task',
    python_callable=log_variable,
    dag=dag,
)
 DAG を実行するDAG を Cloud Composer にデプロイし、手動実行します。

これにより、DAG のパースプロセスが開始され、指定したログが出力されるはずです。
ログを確認してみましょう。

Cloud Composer のコンソール画面の [ログ] をクリックして、Cloud Logging の ログエクスプローラ画面を開きます。
DAG の Airflow 変数のアクセスログのみ表示させたいため、下記クエリでログをフィルタリングします。
resource.type="cloud_composer_environment"
severity=INFO
SEARCH("variable `access:`")
トップレベルとタスク内それぞれの Airflow 変数アクセスログが出力されていることが確認できました。
INFO 2024-11-09 12:05:16.279 Top-level variable access: my_value
INFO 2024-11-09 12:05:17.427 Task-level variable access: my_value
この時点では、DAG の実行時のログは確認できましたが、定期的に実行されるはずの DAG パース時のアクセスログは出力されていませんでした。

これは、DAG パーサーのログが標準出力に向けられていないためでした。
 Airflow 構成オプションをオーバーライドするCloud Composer では、Airflow の構成オプションをオーバーライドすることが可能です。
dag_processor_log_target を stdout に設定することで、ログを標準出力に出力し、Cloud Logging に転送されるようにします。
以下のコマンドでオーバーライドします。
$ gcloud composer environments update test-composer \
    --location asia-northeast1 \
    --update-airflow-configs=logging-dag_processor_log_target=stdout
 DAG パーサーのログを確認する再び Cloud Logging の ログエクスプローラ画面を開き、DAG パーサーのログが出力されているかを確認します。
トップレベルの Airflow 変数アクセスログが複数出力されています。
INFO 2024-11-09 12:15:10.109 Top-level variable access: my_value
INFO 2024-11-09 12:15:43.904 Top-level variable access: my_value
INFO 2024-11-09 12:16:16.508 Top-level variable access: my_value
INFO 2024-11-09 12:16:48.514 Top-level variable access: my_value
INFO 2024-11-09 12:17:20.425 Top-level variable access: my_value
INFO 2024-11-09 12:17:52.194 Top-level variable access: my_value
INFO 2024-11-09 12:18:24.003 Top-level variable access: my_value
INFO 2024-11-09 12:18:55.419 Top-level variable access: my_value
INFO 2024-11-09 12:19:27.600 Top-level variable access: my_value
INFO 2024-11-09 12:19:59.300 Top-level variable access: my_value
ログを確認すると、約30秒間隔で定期的にトップレベルの変数アクセスが行われていることが確認できました。

これは、Cloud Composer の DAG パーサーが定期的に DAG ファイルをスキャンし、パースを実行していることを示しています。
 まとめ以上、Cloud Composer で DAG パーサーのログを Cloud Logging に出力する方法について解説しました。
今回の検証により、DAG パース時の変数アクセスの挙動を実際に確認することができました。

約30秒間隔でトップレベルコードでの変数アクセスが発生していることが確認でき、「トップレベルでの変数アクセスは避けるべき」という Airflow のベストプラクティスの根拠を実際のログで理解することができました。
DAG パーサーのログは、スケジューラプロセスによって生成されるため、スケジューラのログに含まれます。

これらのログには、DAG のパース中に発生したエラーや警告、情報メッセージが記録されます。

Cloud Logging でこれらのログを可視化することで、パフォーマンスの問題や異常な動作の早期発見、トラブルシューティングの効率化につながります。
みなさんのワークフロー管理にも取り入れてみてください！
 参考Configuration Reference — Airflow Documentation
Global environment variables | Astronomer Documentation
Airflow コマンドライン インターフェースにアクセスする  |  Cloud Composer  |  Google Cloud
Airflow 構成オプションをオーバーライドする  |  Cloud Composer  |  Google Cloud
Best Practices — Airflow Documentation
Cloud Composer で Airflow DAG 解析時間を短縮す | Google Cloud 公式ブログ